package com.shujia.flink.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

object Demo1Sink {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val studentDS: DataStream[String] = env.readTextFile("data/students.txt")


    /**
      * 1、打印出来
      *
      */
    studentDS.print()


    //使用自定义sink
    studentDS.addSink(new MySInk)


    env.execute()

  }

}

/**
  * 自定义sink
  * SinkFunction： 并行的
  * RichSinkFunction: 多了open和close方法
  *
  */

class MySInk extends RichSinkFunction[String] {

  var con: Connection = _

  /**
    * 在invoke 方法之前执行，在每一个并行的中只执行一次
    *
    */
  override def open(parameters: Configuration): Unit = {

    println("创建链接")

    //将数据保存到mysql中
    //1、加载驱动
    Class.forName("com.mysql.jdbc.Driver")

    //2、创建链接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=utf-8", "root", "123456")

  }

  /**
    * close方法在invoke之后执行，每一个并行度中只执行一次
    */
  override def close(): Unit = {
    con.close()
  }

  /**
    * invoke： 每一条数据执行一次
    *
    * @param value   : 上游发送过来的数据
    * @param context ： 上下文对象
    */
  override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {

    //编写sql将数据保存到mysql中
    val stat: PreparedStatement = con.prepareStatement("insert into student(id,name,age,gender,clazz) values(?,?,?,?,?)")

    //设置列值
    val split: Array[String] = value.split(",")

    stat.setString(1, split(0))
    stat.setString(2, split(1))
    stat.setInt(3, split(2).toInt)
    stat.setString(4, split(3))
    stat.setString(5, split(4))

    stat.executeUpdate()

  }
}
